/**
 * 
 */
package edu.buffalo.cse.cse486_586.simpledynamo.provider;

import java.io.IOException; 
import java.io.InterruptedIOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;

import edu.buffalo.cse.cse486_586.simpledynamo.vo.MessageObject;
import edu.buffalo.cse.cse486_586.simpledynamo.vo.MessageType;

import android.content.ContentProvider;
import android.content.ContentUris;
import android.content.ContentValues;
import android.database.Cursor;
import android.database.MatrixCursor;
import android.database.sqlite.SQLiteDatabase;
import android.database.sqlite.SQLiteOpenHelper;
import android.database.sqlite.SQLiteQueryBuilder;
import android.net.Uri;
import android.util.Log;

/**
 * @author Sravan
 */
public class MyContentProvider extends ContentProvider{

	public static final String AUTHORITY = "edu.buffalo.cse.cse486_586.simpledynamo.provider";
	public static final Uri CONTENT_URI = Uri.parse("content://" + AUTHORITY);
	private static final String SERVER_ADDRESS = "10.0.2.2";
	private static final int SERVER_BIND_PORT = 10000;
	private static int MAX_NODES_ALLOWED = 5;

	private static int INITIAL_DELAY_AFTER_SERVER_START = 500;

	public static int RESPONSE_TIMEOUT = 1000;
	public static int MULTI_LEVEL_RESPONSE_TIMEOUT = 5 * RESPONSE_TIMEOUT;

	public static String nodeID;
	public static boolean stopNode = false;

	private final String DATABASE_NAME = "KeyValueStore";
	private final String COORDINATOR_FAILED = "COORDINATOR_FAILED";
	private final String __MESSAGE_TYPE__ = "__MESSAGE_TYPE__";
	private final String __NULL__ = "NULL";

	private final String LOG_TAG = "S_DYNAMO";
	private final String QUEUE_LOG_TAG = "QUEUES_S_DYNAMO";

	private SQLiteOpenHelper dataHelper;
	ArrayList<Node> nodesList = new ArrayList<Node>();
	private ArrayBlockingQueue<String> uriBlockingQueue = new ArrayBlockingQueue<String>(5);
	private ArrayBlockingQueue<String> cursorBlockingQueue = new ArrayBlockingQueue<String>(5);
	private ArrayBlockingQueue<String> replicaResponsesBlockingQueue = new ArrayBlockingQueue<String>(5);
	private ArrayBlockingQueue<String> replicaCursorBlockingQueue = new ArrayBlockingQueue<String>(5);

	private final int READ_QUORUM = 2;
	private final int WRITE_QUORUM = 2;
	private final int NO_OF_REPLICAS = 3;
	private Map<String,String> records_onJoin = new HashMap<String, String>();

	public int delete(Uri uri, String selection, String[] selectionArgs) {
		SQLiteDatabase db = dataHelper.getWritableDatabase();

		int no_of_records = db.delete(MyDataHelper.TABLE_NAME, null, null);

		getContext().getContentResolver().notifyChange(uri, null);
		return no_of_records;
	}

	@Override
	public String getType(Uri uri) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Uri insert(Uri uri, ContentValues values) {
		SQLiteDatabase db = dataHelper.getWritableDatabase();

		String key = values.getAsString(MyDataHelper.COLUMN_KEY);
		String value = values.getAsString(MyDataHelper.COLUMN_VALUE);
		String cvType = "";
		if(values.containsKey(__MESSAGE_TYPE__)){
			cvType = values.getAsString(__MESSAGE_TYPE__);
		}

		Node node = getNodeResponsibleForKey(key);
		Log.d(LOG_TAG,"insert(): ["+key + "," +value + "] -- ("+nodeID + ", "+node.node_id + ")");
		if(nodeID.equals(node.node_id) && cvType.equals("")){
			// update if key already exists. 
			// Current Node & I didn't fail. (since I wouldn't be here if I failed)
			/*long id = 0; // i
			if((queryForKeyLocal(key)).getCount() > 0){
				id = update(uri, values, key, null); // expected return value is '1'
			}
			else{
				id = db.insert(MyDataHelper.TABLE_NAME, null, values);
			}*/
			//if(id >= 0){
				// local_success, REPLICATE AT OTHERS
				// get nodes responsibles for key (including replicas)
				Node rep1 = getMy_First_ReplicaNode(nodeID);
				if(rep1 != null){
					MessageObject msg1 = new MessageObject();
					msg1.setKey(key);
					msg1.setValue(value);
					msg1.setNode_id(nodeID);
					msg1.setMessageType(MessageType.REPLICATE_INSERT_REQUEST);
					Log.d(LOG_TAG,"Replicate at --> "+rep1.node_id);
					sendMessageToOtherNode(msg1,Utility.resolvePortNumber(rep1.node_id),RESPONSE_TIMEOUT);
				}
				Node rep2 = getMy_Second_ReplicaNode(nodeID);
				if(rep2 != null){
					MessageObject msg2 = new MessageObject();
					msg2.setKey(key);
					msg2.setValue(value);
					msg2.setNode_id(nodeID);
					msg2.setMessageType(MessageType.REPLICATE_INSERT_REQUEST);
					Log.d(LOG_TAG,"Replicate at "+rep2.node_id);
					sendMessageToOtherNode(msg2,Utility.resolvePortNumber(rep2.node_id),RESPONSE_TIMEOUT);
				}
				// TODO require a blocking queue to collect data.
				// NULLS CAN BE INSERTED INTO BlockingQueue. -- nulls cannot be inserted, instead some strings can be.

				int i=0;
				int write_quorum_reponses = 0;
				for( ;i<NO_OF_REPLICAS-1 && getNoOfNodes() > 1 ;i++){ // Excluding me. 
					try {
						String msg = replicaResponsesBlockingQueue.take();
						Log.d(LOG_TAG, "REPLICATION responses: "+msg);
						if( ! msg.equals(__NULL__) ){
							write_quorum_reponses++;
						}
						if(write_quorum_reponses == WRITE_QUORUM - 1){ // Considering I'm IN
							//break; // TODO break not needed as all responses must be pulled out of queue.
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}				
				}
				if(write_quorum_reponses < WRITE_QUORUM-1){ // including me.
					// TODO delete/revert back data. THIS may not happen.
					Log.d(LOG_TAG, "WRITE_QUORUM NOT achieved.");
					return null;
				}
				
				long id = 0; // i
				if((queryForKeyLocal(key)).getCount() > 0){
					id = update(uri, values, key, null); // expected return value is '1'
				}
				else{
					id = db.insert(MyDataHelper.TABLE_NAME, null, values);
				}

				//replicaResponsesBlockingQueue.clear(); // TODO remove this in final code.

				Uri ret_Uri = ContentUris.withAppendedId(MyContentProvider.CONTENT_URI, id);
				getContext().getContentResolver().notifyChange(ret_Uri, null);
				return ret_Uri;
			//}
			//return null;
		}

		else if(cvType.equals("")){ // Other nodes, first call.

			MessageObject msg = new MessageObject();
			msg.setKey(key);
			msg.setValue(value);
			msg.setNode_id(nodeID);
			msg.setMessageType(MessageType.INSERT_REQUEST);
			// TODO: Handle coordinator failure.
			sendMessageToOtherNode(msg,Utility.resolvePortNumber(node.node_id),MULTI_LEVEL_RESPONSE_TIMEOUT); // handle this timeout specifically.
			try {
				String ret_uri = uriBlockingQueue.take();
				if(ret_uri.equals(COORDINATOR_FAILED)){ //TODO: for loop can be placed here for a generic case.
					node = getMy_First_ReplicaNode(node.node_id);
					msg.setMessageType(MessageType.REPLICA_INSERT_REQUEST);
					sendMessageToOtherNode(msg,Utility.resolvePortNumber(node.node_id),MULTI_LEVEL_RESPONSE_TIMEOUT);
					ret_uri = uriBlockingQueue.take();
				}
				if(ret_uri.equals(COORDINATOR_FAILED)){
					Log.d(LOG_TAG, "Exception: INSERT_RETURN_VALUE_IS_NULL");
					return null;
				}
				return Uri.parse(ret_uri);
			} catch (InterruptedException e) {
				Log.d(LOG_TAG, "Exception: "+e.getMessage());
				return null;
			}
		}
		else{
			values.remove(__MESSAGE_TYPE__);
			if(cvType.equals(MessageType.REPLICATE_INSERT_REQUEST.toString())){
				// JUST INSERT
				long id = 0; // i
				if((queryForKeyLocal(key)).getCount() > 0){
					id = update(uri, values, key, null); // expected return value is '1'
				}
				else{
					id = db.insert(MyDataHelper.TABLE_NAME, null, values);
				}

				Log.d(LOG_TAG,"Replicated key --> "+key);
				if(id >= 0){
					Uri ret_Uri = ContentUris.withAppendedId(MyContentProvider.CONTENT_URI, id);
					getContext().getContentResolver().notifyChange(ret_Uri, null);
					return ret_Uri;
				}
			}
			else if(cvType.equals(MessageType.REPLICA_INSERT_REQUEST.toString())){
				//long id = db.insert(MyDataHelper.TABLE_NAME, null, values);
				//if(id > 1){
					Node rep2 = getMy_First_ReplicaNode(nodeID);
					MessageObject msg1 = new MessageObject();
					msg1.setKey(key);
					msg1.setValue(value);
					msg1.setNode_id(nodeID);
					msg1.setMessageType(MessageType.REPLICATE_INSERT_REQUEST);
					sendMessageToOtherNode(msg1,Utility.resolvePortNumber(rep2.node_id),RESPONSE_TIMEOUT);


					try {
						String msg;
						msg = replicaResponsesBlockingQueue.take();
						if(msg.equals(__NULL__)){ 
							return null; // TODO update replicas first, and then update mine. (easy to achieve consistancy)
							//insert failed, revert my insert,
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					long id = 0; // i
					if((queryForKeyLocal(key)).getCount() > 0){
						id = update(uri, values, key, null); // expected return value is '1'
					}
					else{
						id = db.insert(MyDataHelper.TABLE_NAME, null, values);
					}
					
					//long id = db.insert(MyDataHelper.TABLE_NAME, null, values);
					Uri ret_Uri = ContentUris.withAppendedId(MyContentProvider.CONTENT_URI, id);
					getContext().getContentResolver().notifyChange(ret_Uri, null);
					return ret_Uri;
				//}
				//return null;
			}
		}

		return null;

	}

	private void updateDataIn_newNode(Map<String, String> records){
		/*if(nodesList.size() != 5){
			return;
		}*/
		Log.d(QUEUE_LOG_TAG, "updateDataIn_newNode() BEGIN: no_of_recs="+records.size());
		/*if(records.size() > 0)
			delete(CONTENT_URI, null, null); // DELETE ALL RECORDS
		*/
		SQLiteDatabase db = dataHelper.getWritableDatabase();
		for(String key:records.keySet()){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, key);
			cv.put(MyDataHelper.COLUMN_VALUE, records.get(key));
			long id = 0; // i
			if((queryForKeyLocal(key)).getCount() > 0){
				id = update(CONTENT_URI, cv, key, null); // expected return value is '1'
			}
			else{
				id = db.insert(MyDataHelper.TABLE_NAME, null, cv);
			}
		}
		
		Log.d(QUEUE_LOG_TAG, "updateDataIn_newNode() END");
		
	}
	
	private Map<String,String> getKeysForNode_onJoin(String node_id){
		
		Log.d(QUEUE_LOG_TAG, "getKeysForNode_onJoin(): for "+node_id); 
		
		Map<String, String> ret_val = new HashMap<String, String>();
		/*if(nodesList.size() != 5){
			return ret_val;
		}*/
		
		Cursor cursor = queryForKeyLocal(null);
		// TODO calculate hashcode of starting value.
		int idx = 0;
		for(Node node:nodesList){
			if(node.node_id.equals(node_id)){		
				break;
			}
			idx++;
		}
		// required node = (idx-2)
		int idx1 = (idx-3+nodesList.size())%nodesList.size();
		String endHashValue = nodesList.get(idx).hashValue;
		String startHashValue = nodesList.get(idx1).hashValue;
		int count = 0;
		while(cursor.moveToNext()){
			count++;
			// add only required keys to list.
			if(isHashBetweenHashes(startHashValue,endHashValue,Utility.genHash(cursor.getString(0)))){
				Log.d(QUEUE_LOG_TAG,cursor.getString(0)+" added");
				ret_val.put(cursor.getString(0), cursor.getString(1));
			}
		}
		Log.d(QUEUE_LOG_TAG, "getKeysForNode_onJoin(): total_keys=" + count + ", -- keys sent="+ret_val.size()); 
		return ret_val;
	}
	
	private boolean isHashBetweenHashes(String start, String end, String find){
		Log.d(QUEUE_LOG_TAG, start+" : "+find + " : "+end);
		if(start.compareTo(end) < 0){
			if(start.compareTo(find) <0 && end.compareTo(find) > 0){
				return true;
			}
		}
		else if(start.compareTo(end) > 0){
			if(start.compareTo(find) < 0 || end.compareTo(find) > 0){
				return true;
			}
		}
		return false;
	}
	
	@Override
	public boolean onCreate() {
		dataHelper = new MyDataHelper(getContext(),
				DATABASE_NAME,
				null,
				1);

		// Instead of calling init directly, introduce a delay, onCreate must be finished
		(new Thread(new InitRunnable())).start();
		Log.d(LOG_TAG,"MCP: "+"MyContentProvider.onCreate(): Begin");
		//return false;
		return true;
	}

	@Override
	public Cursor query(Uri uri, String[] projection, String selection,
			String[] selectionArgs, String sortOrder) {

		if(sortOrder != null){
			Log.d(QUEUE_LOG_TAG,selection + " : "+sortOrder);
		}
		
		SQLiteQueryBuilder qBuilder = new SQLiteQueryBuilder();
		qBuilder.setTables(MyDataHelper.TABLE_NAME);
		SQLiteDatabase db = dataHelper.getReadableDatabase();
		if(selection == null){ // to handle data dumps from local.
			Cursor cursor = qBuilder.query(db, projection, selection , null, null, null, null);
			cursor.setNotificationUri(getContext().getContentResolver(), uri);
			return cursor;
		}
		
		String queryType = "";
		if(sortOrder != null/* && sortOrder.equals(MessageType.QUERY_REPLICATE_REQUEST.toString())*/){
			queryType = sortOrder;
		}

		//TODO handle selection = null to fetch local dump.

		String key = selection+"";
		Node node = getNodeResponsibleForKey(key);

		if(nodeID.equals(node.node_id) && queryType.equals("")){
			// TODO contact replica other nodes.
			Node rep1 = getMy_First_ReplicaNode(nodeID);
			int count = 0;
			if(rep1 != null){
				MessageObject msg1 = new MessageObject();
				msg1.setKey(key);
				msg1.setNode_id(nodeID);
				msg1.setMessageType(MessageType.QUERY_REPLICATE_REQUEST);
				Log.d(LOG_TAG,"Query request at --> " + rep1.node_id);
				sendMessageToOtherNode(msg1,Utility.resolvePortNumber(rep1.node_id),RESPONSE_TIMEOUT);
				count++;
				//TODO receive value in the blocking queue for the key.
			}
			Node rep2 = getMy_Second_ReplicaNode(nodeID);
			if(rep2 != null){
				MessageObject msg2 = new MessageObject();
				msg2.setKey(key);
				msg2.setNode_id(nodeID);
				msg2.setMessageType(MessageType.QUERY_REPLICATE_REQUEST);
				Log.d(LOG_TAG,"Query request at --> " + rep2.node_id);
				sendMessageToOtherNode(msg2,Utility.resolvePortNumber(rep2.node_id),RESPONSE_TIMEOUT);
				count++;
			}
			// TODO require a blocking queue to collect value.
			// NULLS CAN BE INSERTED INTO BlockingQueue. -- nulls cannot be inserted, instead some strings can be.

			int i=0;
			int read_quorum_reponses = 0;
			for( ;i<count && getNoOfNodes() > 1 ;i++){ // Excluding me. 
				try {
					String msg = replicaCursorBlockingQueue.take();
					Log.d(QUEUE_LOG_TAG,"TAKE -- replicaCursorBlockingQueue: ("+key+")");
					Log.d(LOG_TAG, "REPLICATION responses: "+msg);
					if( ! msg.equals(__NULL__) ){
						read_quorum_reponses++;
					}
					if(read_quorum_reponses == READ_QUORUM - 1){ // Considering I'm IN
						//break; // TODO break not needed as all responses must be pulled out of queue.
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}				
			}
			if(read_quorum_reponses < READ_QUORUM-1){ // including me.
				// TODO delete/revert back data. THIS may not happen.
				Log.d(LOG_TAG, "WRITE_QUORUM NOT achieved.");
				MatrixCursor cursor = new MatrixCursor(new String[]{MyDataHelper.COLUMN_KEY, MyDataHelper.COLUMN_VALUE});
				return cursor; // TODO return empty matrix cursor.
			}

			//replicaResponsesBlockingQueue.clear(); // TODO remove this in final code.

			selection = MyDataHelper.COLUMN_KEY + " = '" + selection + "'";
			Cursor cursor = qBuilder.query(db, projection, selection , null, null, null, null);
			cursor.setNotificationUri(getContext().getContentResolver(), uri);

			return cursor;

		}
		//------------------------------------------
		else if(queryType.equals("")){
			MessageObject msg = new MessageObject();
			msg.setKey(key);
			msg.setNode_id(nodeID);
			msg.setMessageType(MessageType.QUERY_REQUEST);
			// TODO: Handle coordinator failure.
			sendMessageToOtherNode(msg,Utility.resolvePortNumber(node.node_id),MULTI_LEVEL_RESPONSE_TIMEOUT); // handle this timeout specifically.
			try {
				String ret_uri = cursorBlockingQueue.take();
				Log.d(QUEUE_LOG_TAG,"TAKE -- cursorBlockingQueue: ("+key+")");
				MatrixCursor cursor = new MatrixCursor(new String[]{MyDataHelper.COLUMN_KEY, MyDataHelper.COLUMN_VALUE});
				if(ret_uri.equals(COORDINATOR_FAILED)){
					// TODO coordinator failed.
					Log.d(LOG_TAG, "Coordinator " + node.node_id + " failed for key("+key+")");
					node = getMy_First_ReplicaNode(node.node_id);
					Log.d(LOG_TAG, "New Coordinator is: " + node.node_id);
					msg.setMessageType(MessageType.QUERY_REPLICA_REQUEST);
					sendMessageToOtherNode(msg,Utility.resolvePortNumber(node.node_id),MULTI_LEVEL_RESPONSE_TIMEOUT);
					Log.d(LOG_TAG, "RESPONSE from back-up co-ord");
					ret_uri = cursorBlockingQueue.take();
					Log.d(QUEUE_LOG_TAG,"TAKE -- cursorBlockingQueue: ("+key+")");
					if(ret_uri.equals(COORDINATOR_FAILED)){
						return cursor;
					}					
				}
				cursor.addRow(new String[]{key,ret_uri});
				return cursor;

				/*if(ret_uri.equals(COORDINATOR_FAILED)){ //TODO: for loop can be placed here for a generic case.
					node = getMy_First_ReplicaNode(node.node_id);
					msg.setMessageType(MessageType.REPLICA_INSERT_REQUEST);
					sendMessageToOtherNode(msg,Utility.resolvePortNumber(node.node_id),MULTI_LEVEL_RESPONSE_TIMEOUT);
					ret_uri = uriBlockingQueue.take();
				}
				if(ret_uri.equals(__NULL__)){
					Log.d(LOG_TAG, "Exception: INSERT_RETURN_VALUE_IS_NULL");
					return null;
				}*/
				//return Uri.parse(ret_uri);
			} catch (InterruptedException e) {
				Log.d(LOG_TAG, "Exception: "+e.getMessage());
				return null;
			}
		}
		else{
			if(queryType.equals(MessageType.QUERY_REPLICATE_REQUEST.toString())){
				Log.d(LOG_TAG,"Queried at replica for key: ("+selection+")");
				selection = MyDataHelper.COLUMN_KEY + " = '" + selection + "'";
				Cursor cursor = qBuilder.query(db, projection, selection , null, null, null, null);
				cursor.setNotificationUri(getContext().getContentResolver(), uri);
				return cursor;
			}
			else if(queryType.equals(MessageType.QUERY_REPLICA_REQUEST.toString())){
				Log.d(LOG_TAG,"Query at REPLICA Co_ord --> " + nodeID);
				Node rep1 = getMy_First_ReplicaNode(nodeID);
				if(rep1 != null){
					MessageObject msg1 = new MessageObject();
					msg1.setKey(key);
					msg1.setNode_id(nodeID);
					msg1.setMessageType(MessageType.QUERY_REPLICATE_REQUEST);
					Log.d(LOG_TAG,"Query request at --> " + rep1.node_id);
					sendMessageToOtherNode(msg1,Utility.resolvePortNumber(rep1.node_id),RESPONSE_TIMEOUT);
					//TODO receive value in the blocking queue for the key.
					try {
						MatrixCursor cursor = new MatrixCursor(new String[]{MyDataHelper.COLUMN_KEY, MyDataHelper.COLUMN_VALUE});
						String value = replicaCursorBlockingQueue.take();
						Log.d(QUEUE_LOG_TAG,"TAKE -- replicaCursorBlockingQueue: ("+key+")");
						Log.d(LOG_TAG, "REPLICATION responses: "+value);
						if( value.equals(__NULL__) ){
							return cursor;
						}
						selection = MyDataHelper.COLUMN_KEY + " = '" + selection + "'";
						Cursor cursor1 = qBuilder.query(db, projection, selection , null, null, null, null);
						cursor.setNotificationUri(getContext().getContentResolver(), uri);
						if(cursor1.moveToFirst()){
							if(value.equals(cursor1.getString(1))){
								cursor.addRow(new String[]{key,value});
							}
						}
						return cursor;
						
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			}
		}
		return null;

	}

	private Cursor queryForKeyLocal(String selection){
		SQLiteQueryBuilder qBuilder = new SQLiteQueryBuilder();
		qBuilder.setTables(MyDataHelper.TABLE_NAME);
		if(selection != null){
			//qBuilder.appendWhere("provider_key = " + selection + "");
			selection = MyDataHelper.COLUMN_KEY + " = '" + selection + "'";
		}

		SQLiteDatabase db = dataHelper.getReadableDatabase();

		//Cursor cursor = qBuilder.query(db, projection, "provider_key = '" + selection + "'", null, null, null, null);
		Cursor cursor = qBuilder.query(db, null, selection , null, null, null, null);
		cursor.setNotificationUri(getContext().getContentResolver(), MyContentProvider.CONTENT_URI);

		return cursor;
	}

	@Override
	public int update(Uri uri, ContentValues values, String selection,
			String[] selectionArgs) {
		SQLiteDatabase db = dataHelper.getWritableDatabase();

		if(selection == null){
			return 0;
		}
		else{
			selection = MyDataHelper.COLUMN_KEY + " = '" + selection + "'";
		}

		int id = db.update(MyDataHelper.TABLE_NAME, values, selection, null);
		getContext().getContentResolver().notifyChange(uri, null);

		return id;
	}

	private synchronized void sendMessageToOtherNode(MessageObject msgObj, int port_number, int delay){
		// Intentionally created new objects.
		//String key = new String(l_key);
		//String value = new String(l_value);
		//if(!nodeID.equals(CENTRAL_SERVER_ID)){ // If not 5554. (New Node JOIN.)
		(new Thread(new ClientRunnable(msgObj, port_number, delay))).start();
		//}
	}

	public synchronized void newNodeJoined_client(MessageObject msg){
		String newHashValue = Utility.genHash(msg.getNode_id());
		int index = 0; 
		if(msg.getRecords() != null){
			Log.d(QUEUE_LOG_TAG, "records: "+msg.getRecords());
			//records_onJoin.putAll(msg.getRecords()); // INSERT MUST HAPPEN.
			updateDataIn_newNode(msg.getRecords());
		}
		for(Node node:nodesList){ // If already present, don't insert.
			if(node.node_id.equals(msg.getNode_id())){
				//records_onJoin.putAll(msg.getRecords());
				return;
			}
		}

		for(Node node:nodesList){
			if(Utility.hashCompare(node.hashValue, newHashValue) > 0){
				break;
			}
			index++;
		}
		nodesList.add(index, new Node(msg.getNode_id(),newHashValue)); // Add node at the correct position.

		String nodesListTemp = "";
		String nodesListHashValue = "";
		for(Node node:nodesList){
			nodesListTemp+= node.node_id + ", ";
			nodesListHashValue += node.hashValue + ", ";
		}
		Log.d(LOG_TAG, nodesListTemp);
		Log.d(QUEUE_LOG_TAG, nodesListHashValue);
		
		//updateDataIn_newNode(msg.getRecords());
		
	}
	
	public synchronized void newNodeJoined_server(MessageObject msg){
		String newHashValue = Utility.genHash(msg.getNode_id());
		int index = 0; 
		
		Log.d(QUEUE_LOG_TAG, "newNodeJoined_server(): "+msg.getNode_id());
		
		for(Node node:nodesList){ // If already present, don't insert.
			if(node.node_id.equals(msg.getNode_id())){
				msg.setRecords(getKeysForNode_onJoin(node.node_id));
				return;
			}
		}

		for(Node node:nodesList){
			if(Utility.hashCompare(node.hashValue, newHashValue) > 0){
				break;
			}
			index++;
		}
		nodesList.add(index, new Node(msg.getNode_id(),newHashValue)); // Add node at the correct position.

		String nodesListTemp = "";
		for(Node node:nodesList)
			nodesListTemp+= node.node_id + ", ";
		Log.d(LOG_TAG, nodesListTemp);
		
		//updateDataIn_newNode(msg.getRecords());
		//records_onJoin.putAll(msg.getRecords());
		
	}

	private synchronized int getNoOfNodes(){
		return nodesList.size();
	}

	private synchronized Node getMy_First_ReplicaNode(String node_Id){
		if(nodesList.size() == 1){
			return null;
		}
		//String hashedKey = Utility.genHash(key);
		int index = 0;
		for(Node node:nodesList){
			if(node.node_id.compareToIgnoreCase(node_Id) == 0){
				break;
			}
			index++;
		}
		if(index == nodesList.size()){
			return null;
		}
		return nodesList.get( (index+1)%nodesList.size() ); // overFlow, ffff belongs to node '000a' etc
	}

	private synchronized Node getMy_Second_ReplicaNode(String node_Id){
		if(nodesList.size() <= 2){
			return null;
		}
		//String hashedKey = Utility.genHash(key);
		int index = 0;
		for(Node node:nodesList){
			if(node.node_id.compareToIgnoreCase(node_Id) == 0){
				break;
			}
			index++;
		}
		if(index == nodesList.size()){
			Log.d(LOG_TAG, "CANNOT HAPPEN");
			return null; // my node not in list, surprising!!
		}
		return nodesList.get( (index+2)%nodesList.size() ); // overFlow, ffff belongs to node '000a' etc
	}

	/**
	 * Node Information
	 * @author sravan
	 */
	private class Node{
		public String node_id;
		public String hashValue;
		public Node(String node_id, String hashValue){
			this.node_id = node_id;
			this.hashValue = hashValue;
		}
	}	

	private Node getNodeResponsibleForKey(String key){
		String hashedKey = Utility.genHash(key);
		for(Node node:nodesList){
			if(node.hashValue.compareToIgnoreCase(hashedKey) >= 0){
				return node;
			}
		}
		return nodesList.get(0); // overFlow, ffff belongs to node '000a' etc
	}

	private int getServerPort(){
		return SERVER_BIND_PORT;
	}

	private class InitRunnable implements Runnable{
		@Override
		public void run() {
			try{
				Thread.sleep(2000);
			}
			catch(Exception e){
				e.printStackTrace();
			}
			init();
		}

	}

	private void init() {
		// Every node will have a server.

		Log.d(LOG_TAG,"INIT: "+nodeID);
		try{
			Thread serverThread = new Thread(new ServerRunnable(getServerPort(), nodeID));
			serverThread.start();
			Thread.sleep(INITIAL_DELAY_AFTER_SERVER_START); // allow for some intial delay.
		}
		catch(InterruptedException e){
			e.printStackTrace();
		}
		int myPortNumber = Utility.resolvePortNumber(nodeID);

		nodesList.add(new Node(nodeID, Utility.genHash(nodeID))); // ADD my node to nodesList.

		for(int i=0;i<MAX_NODES_ALLOWED;i++){
			int t_port_no = (5554 + 2*i) * 2; 
			if(t_port_no != myPortNumber){ // Don't connect to mine. but instead, add my node data to nodesList.
				MessageObject msg = new MessageObject();
				msg.setNode_id(nodeID);
				msg.setMessageType(MessageType.NEW_NODE_REQUEST);

				(new Thread( new ClientRunnable(msg, t_port_no, RESPONSE_TIMEOUT*3))).start();
			}
		}
		/*try{
			Thread.sleep(5000);
		}
		catch(InterruptedException e){
			e.printStackTrace();
		}
		Log.d(QUEUE_LOG_TAG,"init():records: "+records_onJoin);
		updateDataIn_newNode(records_onJoin);
		records_onJoin.clear();*/
	}

	private void readTimeoutOccurred(MessageType msgType){
		if(msgType == MessageType.INSERT_REQUEST){
			//insert failed, as no response from co-ordinator
			uriBlockingQueue.add(COORDINATOR_FAILED);
		}
		else if(msgType == MessageType.REPLICATE_INSERT_REQUEST){
			try {
				replicaResponsesBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msgType == MessageType.REPLICA_INSERT_REQUEST){
			try {
				uriBlockingQueue.put(COORDINATOR_FAILED);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msgType == MessageType.QUERY_REQUEST){
			//insert failed, as no response from co-ordinator
			cursorBlockingQueue.add(COORDINATOR_FAILED);
			Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+COORDINATOR_FAILED+")");
		}
		else if(msgType == MessageType.QUERY_REPLICA_REQUEST){
			Log.d(LOG_TAG, "Time out for response.");
			cursorBlockingQueue.add(COORDINATOR_FAILED);
			Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+COORDINATOR_FAILED+")");
		}
		else if(msgType == MessageType.QUERY_REPLICATE_REQUEST){
			try {
				replicaCursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- replicaCursorBlockingQueue: ("+__NULL__+")");				
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void messageFromOtherNode(MessageObject msg) {
		//String src_node_id = msg.getNode_id()+""; //
		if(msg.getMessageType() == MessageType.INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);

			if(ret_uri == null){ // Insert failed
				msg.setMessageType(MessageType.INSERT_FAILED);
				msg.setNode_id(nodeID);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{ // Insert Success
				msg.setMessageType(MessageType.INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				uriBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.INSERT_SUCCESS){
			try {
				uriBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			cv.put(__MESSAGE_TYPE__,MessageType.REPLICATE_INSERT_REQUEST.toString());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);

			if(ret_uri == null){ // Insert failed
				Log.d(LOG_TAG, " not_inserted at replica?? uri is NULL");
				msg.setMessageType(MessageType.REPLICATE_INSERT_FAILED);
				msg.setNode_id(nodeID);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),RESPONSE_TIMEOUT);
			}
			else{ // Insert Success
				Log.d(LOG_TAG, "Inserted at replica: uri is "+ret_uri);
				msg.setMessageType(MessageType.REPLICATE_INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),RESPONSE_TIMEOUT);
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				replicaResponsesBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_SUCCESS){
			Log.d(LOG_TAG,"Add to replicaResponsesBlockingQueue() : REPLICATE_INSERT_SUCCESS");
			try {
				replicaResponsesBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			cv.put(__MESSAGE_TYPE__,MessageType.REPLICA_INSERT_REQUEST.toString());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);
			if(ret_uri == null){ // Insert failed
				msg.setMessageType(MessageType.REPLICA_INSERT_FAILED);
				msg.setNode_id(nodeID);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{ // Insert Success
				msg.setMessageType(MessageType.REPLICA_INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		} 
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				uriBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_SUCCESS){
			try {
				uriBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REQUEST){
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, null);
			Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_SUCCESS){
			try {
				cursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_FAILED){
			try {
				cursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + __NULL__+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if (msg.getMessageType() == MessageType.QUERY_REPLICATE_REQUEST){
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, MessageType.QUERY_REPLICATE_REQUEST.toString());
			//Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_REPLICATE_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_REPLICATE_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICATE_SUCCESS){
			try {
				replicaCursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- replicaCursorBlockingQueue: ("+msg.getKey() + ":"+msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICATE_FAILED){
			try {
				replicaCursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- replicaCursorBlockingQueue: ("+msg.getKey() + ":"+__NULL__+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if (msg.getMessageType() == MessageType.QUERY_REPLICA_REQUEST){
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, MessageType.QUERY_REPLICA_REQUEST.toString());
			//Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_REPLICA_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_REPLICA_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICA_SUCCESS){
			try {
				cursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICA_FAILED){
			try {
				cursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + __NULL__ +")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}


private class MessageFromOtherNode implements Runnable{
	
	MessageObject msg;
	Socket client;
	
	public MessageFromOtherNode(MessageObject msg, Socket client) throws IOException {
		this.msg = msg;
		this.client = client;
	}
	
	@Override
	public void run() {
		Log.d(LOG_TAG,msg.toString());
		String src_node_id = msg.getNode_id()+""; //
		if(msg.getMessageType() == MessageType.INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);

			if(ret_uri == null){ // Insert failed
				msg.setMessageType(MessageType.INSERT_FAILED);
				msg.setNode_id(nodeID);
				//(new ObjectOutputStream(client.getOutputStream())).writeObject(msg);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
			else{ // Insert Success
				msg.setMessageType(MessageType.INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
		}
		else if(msg.getMessageType() == MessageType.INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				uriBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.INSERT_SUCCESS){
			try {
				uriBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			cv.put(__MESSAGE_TYPE__,MessageType.REPLICATE_INSERT_REQUEST.toString());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);
			if(ret_uri == null){ // Insert failed
				Log.d(LOG_TAG, "Not_inserted at replica?? uri is NULL");
				msg.setMessageType(MessageType.REPLICATE_INSERT_FAILED);
				msg.setNode_id(nodeID);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
			else{ // Insert Success
				Log.d(LOG_TAG, " inserted at replica: uri is "+ret_uri);
				msg.setMessageType(MessageType.REPLICATE_INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				replicaResponsesBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICATE_INSERT_SUCCESS){
			try {
				replicaResponsesBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_REQUEST){
			ContentValues cv = new ContentValues();
			cv.put(MyDataHelper.COLUMN_KEY, msg.getKey());
			cv.put(MyDataHelper.COLUMN_VALUE, msg.getValue());
			cv.put(__MESSAGE_TYPE__,MessageType.REPLICA_INSERT_REQUEST.toString());
			Uri ret_uri = insert(MyContentProvider.CONTENT_URI,cv);
			if(ret_uri == null){ // Insert failed

				msg.setMessageType(MessageType.REPLICA_INSERT_FAILED);
				msg.setNode_id(nodeID);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
			else{ // Insert Success
				msg.setMessageType(MessageType.REPLICA_INSERT_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setUri(ret_uri.toString());
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(src_node_id),0);
			}
		} 
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_FAILED){ // TODO this and below condition can be merged.
			// THIS CASE DOESN'T ARISE
			try {
				replicaResponsesBlockingQueue.put(__NULL__);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.REPLICA_INSERT_SUCCESS){
			try {
				replicaResponsesBlockingQueue.put(msg.getUri());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REQUEST){
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, null);
			Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_SUCCESS){
			try {
				cursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_FAILED){
			try {
				cursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + __NULL__ +")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if (msg.getMessageType() == MessageType.QUERY_REPLICATE_REQUEST){
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, MessageType.QUERY_REPLICATE_REQUEST.toString());
			//Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_REPLICATE_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_REPLICATE_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICATE_SUCCESS){
			try {
				replicaCursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- replicaCursorBlockingQueue: ("+msg.getKey() + ":"+msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICATE_FAILED){
			try {
				replicaCursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- replicaCursorBlockingQueue: ("+msg.getKey() + ":"+__NULL__+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if (msg.getMessageType() == MessageType.QUERY_REPLICA_REQUEST){
			Log.d(QUEUE_LOG_TAG,"QUERY_REPLICA_REQUEST received");
			Cursor cursor = query(MyContentProvider.CONTENT_URI, null, msg.getKey(), null, MessageType.QUERY_REPLICA_REQUEST.toString());
			//Map<String, String> records = new HashMap<String, String>();
			if(cursor != null && cursor.moveToFirst()){
				msg.setMessageType(MessageType.QUERY_REPLICA_SUCCESS);
				msg.setNode_id(nodeID);
				msg.setValue(cursor.getString(1));
				Log.d("DATA",cursor.getString(0) + ":" + cursor.getString(1));
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
			else{
				msg.setMessageType(MessageType.QUERY_REPLICA_FAILED);
				msg.setNode_id(nodeID);
				msg.setValue(null);
				//sendMessageToOtherNode(msg,Utility.resolvePortNumber(msg.getNode_id()),0);
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICA_SUCCESS){
			try {
				cursorBlockingQueue.put(msg.getValue());
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + msg.getValue()+")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		else if(msg.getMessageType() == MessageType.QUERY_REPLICA_FAILED){
			try {
				cursorBlockingQueue.put(__NULL__);
				Log.d(QUEUE_LOG_TAG,"PUT -- cursorBlockingQueue: ("+msg.getKey() +":" + __NULL__ +")");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		try {
			(new ObjectOutputStream(client.getOutputStream())).writeObject(msg);
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
		
}

	/**
	 * @author sravan
	 *
	 */
	public class ClientRunnable implements Runnable{

		private String ip_addr = MyContentProvider.SERVER_ADDRESS;
		private int port_number;
		private Socket client;

		private String node_id = MyContentProvider.nodeID;
		private MessageType messageType;
		private MessageObject msgObj;
		private String uri;
		private Map<String, String> records;
		private int delay;

		public ClientRunnable(MessageObject msgObj, int port_number, int delay){
			this.msgObj = msgObj;
			this.port_number = port_number;
			this.delay = delay;
		}

		public void run(){
			try {
				InetAddress addr = InetAddress.getByName(ip_addr);
				client = new Socket(addr, port_number);

				Log.d(LOG_TAG,"CR: "+"Connected to Server!!");
				(new ObjectOutputStream(client.getOutputStream())).writeObject(msgObj);

				try {

					//Log.d(LOG_TAG,"SR: "+"Setting Read Time out");

					client.setSoTimeout(delay);
					ObjectInputStream ois = new ObjectInputStream(client.getInputStream());
					//Log.d(LOG_TAG,"SR: "+"After getting input stream object.");
					MessageType msgType = msgObj.getMessageType();
					MessageObject msg = null;
					msg = (MessageObject)ois.readObject();
					
					if(msg.getKey()!=null && msg.getKey().equals("SYSTEM_DOWN")){ // TODO delete this.
						readTimeoutOccurred(msgObj.getMessageType());
						return;
					}
					
					Log.d(LOG_TAG,"CR: "+msg.toString());
					Log.d(LOG_TAG,"CR: "+msgType+","+msg.getMessageType());
					if(msg.getMessageType() == MessageType.NEW_NODE_SUCCESS){
						newNodeJoined_client(msg);
					}

					else if(msgType != msg.getMessageType()){ // same msg received, ==> just processing at server.
						MessageObject msg1 = new MessageObject();
						msg1.setKey(msg.getKey());
						msg1.setMessageType(msg.getMessageType());
						msg1.setNode_id(msg.getNode_id());
						msg1.setUri(msg.getUri());
						msg1.setValue(msg.getValue());
						messageFromOtherNode(msg1);
					}

				}catch(InterruptedIOException e){
					readTimeoutOccurred(msgObj.getMessageType());
					Log.d(LOG_TAG,"EXCEPTION: "+"ClientRunnable.run(): Read Timeout, "+e.getMessage() + ", timeout delay : "+delay);
					//e.printStackTrace();
				}
				catch (ClassNotFoundException e) {
					Log.d(LOG_TAG,"EXCEPTION: "+"ClientRunnable.run(): ClassNotFoundException "+e.getMessage());
					//e.printStackTrace();
				}

			} catch (IOException e) {
				readTimeoutOccurred(msgObj.getMessageType());
				Log.d(LOG_TAG,"EXCEPTION: "+"Error creating client socket: "+e.getMessage());
				e.printStackTrace();
			}

		}

	}

	private class ServerRunnable implements Runnable {

		private int portNumber = 10000;
		private ServerSocket server = null;
		private Socket client = null;
		private String nodeId = null;

		public ServerRunnable(int i_portNumber,String s_nodeId){
			this.portNumber = i_portNumber;
			this.nodeId = s_nodeId;
		}

		/**
		 * 
		 * (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override 
		public void run() {
			try{
				boolean isStopped = false;
				server = new ServerSocket(portNumber);

				Log.d(LOG_TAG,"SR: "+ "Server at " + nodeId + " binded to port " + portNumber);
				while(true){
					try {
						
						
						client = server.accept();
						
						if(stopNode){ //TODO delete this code.
							isStopped = true;
							try {
								MessageObject msg = (MessageObject) (new ObjectInputStream(client.getInputStream())).readObject();
								msg.setKey("SYSTEM_DOWN");
								(new ObjectOutputStream(client.getOutputStream())).writeObject(msg);
								continue;
							} catch (ClassNotFoundException e) {
								e.printStackTrace();
							}
						}
						
						while(stopNode){
							
							isStopped = true;
							try {
								Thread.sleep(1000);
							} catch (InterruptedException e) {
								e.printStackTrace();
							}
						}
						if(isStopped){ // On resume accept new client.
							
							for(int i=0;i<MAX_NODES_ALLOWED;i++){
								int t_port_no = (5554 + 2*i) * 2; 
								if(t_port_no != Utility.resolvePortNumber(nodeID)){ // Don't connect to mine. but instead, add my node data to nodesList.
									MessageObject msg = new MessageObject();
									msg.setNode_id(nodeID);
									msg.setMessageType(MessageType.NEW_NODE_REQUEST);
									(new Thread( new ClientRunnable(msg, t_port_no, RESPONSE_TIMEOUT))).start();
								}
							}
							
							isStopped = false;
							continue;
						}
						
						Log.d(LOG_TAG,"SR: "+"Client Connected!!");

						// Read a message immediately after connection.
						ObjectInputStream ois = new ObjectInputStream(client.getInputStream());
						MessageObject msg = null;
						try {
							msg = (MessageObject)ois.readObject();
							Log.d(LOG_TAG,"SR: "+msg.toString());
							if( msg.getMessageType() == MessageType.NEW_NODE_REQUEST){ //check if new_node
								Log.d(LOG_TAG,"SR: "+"Calling: newClientSocket() ");

								newNodeJoined_server(msg);
								// TODO inform the system.

								Log.d(QUEUE_LOG_TAG,"records : "+msg.getRecords());
								
								msg.setMessageType(MessageType.NEW_NODE_SUCCESS);
								//msg.setRecords(getKeysForNode_onJoin(msg.getNode_id()));
								msg.setNode_id(nodeID);
								(new ObjectOutputStream(client.getOutputStream())).writeObject(msg);
							}
							else {
								(new Thread(new MessageFromOtherNode(msg, client))).start();
								//messageFromOtherNode(msg,client);
								//(new ObjectOutputStream(client.getOutputStream())).writeObject(msg); // msg modified.
							}

						} catch (ClassNotFoundException e) {
							e.printStackTrace();
							Log.d(LOG_TAG,"EXCEPTION: "+"ServerRunnable.run(): "+e.getMessage());
						}
					}
					catch(IOException e){
						Log.d(LOG_TAG,"SR: " + "Server generated an exception");
						e.printStackTrace();
					}
				}
			} catch (IOException e) {
				Log.d(LOG_TAG,"SR: "+"Error Creating Socket.");
				e.printStackTrace();
			}
		}
	}

}
